package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Demo9_TemporalJoin {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表,kafka中的识别结果数据
        //{"uid":1,"face_res":1,"ts":26000}
        //{"uid":2,"face_res":0,"ts":27000}
        tenv.executeSql("CREATE TABLE face_res              (  \n" +
                "  uid INT                                 \n" +
                "  ,face_res int                              \n" +
                "  ,ts bigint                                 \n" +
                "  ,rt as to_timestamp_ltz(ts,3)             \n" +
                "  ,watermark for rt as rt                   \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'ss-2',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");

        // 建表，映射mysql中的用户最后识别行为事件表 cdc连接器
        tenv.executeSql(
                " create table face_cdc(                    "+
                        "    uid int,                        "+
                        "    event_id string,                "+
                        "    last_action_time bigint,        "+
                        "    rt as to_timestamp_ltz(last_action_time,3) ,     "+
                        "    watermark for rt as rt ,       "+
                        "    primary key(uid) not enforced   "+
                        " ) WITH (                           "+
                        "     'connector' = 'mysql-cdc',     "+
                        "     'hostname' = 'doitedu',        "+
                        "     'port' = '3306',               "+
                        "     'username' = 'root',           "+
                        "     'password' = 'root',           "+
                        "     'database-name' = 'doit47',    "+
                        "     'table-name' = 'last_face_rec' "+
                        " )                                  "
        );


        //tenv.executeSql("select * from face_cdc").print();


        // sql
        tenv.executeSql(
                         " SELECT                                            "+
                        "                                                    "+
                        " a.*,                                               "+
                        " b.*                                                "+
                        "                                                    "+
                        " from face_res as  a                                "+
                        " join face_cdc for system_time as of a.rt as  b     "+
                        "                                                    "+
                        " on a.uid = b.uid                                   "
        ).print();







    }
}
